package com.pajk.user.service.credit;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import javax.annotation.Resource;

import com.alibaba.fastjson.JSON;
import com.pajk.user.dal.mapper.RedoMonitorMapper;
import com.pajk.user.dal.mapper.UserMapper;
import com.pajk.user.model.RedoMonitor;
import com.pajk.user.service.common.idpool.IDPool;
import com.pajk.user.service.common.metaq.MsgObj;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.pajk.user.dal.mapper.CreditHisMapper;
import com.pajk.user.dal.model.CreditHisDO;
import com.pajk.user.service.common.metaq.BaseConsumer;
import com.pajk.user.service.common.metaq.TopicEnum;

/**
 * User: haihua.chenhh
 * Date: 14-3-20
 * Time: 上午11:55
 */
public class CreditAddConsumer extends BaseConsumer {
	
	private static final Logger log = LoggerFactory.getLogger(SynWLTCreditConsumer.class);

    @Resource
    private UserMapper userMapper;

    @Resource
	private CreditHisMapper creditHisMapper;

    @Resource
    RedoMonitorMapper creditErrorMapper;
	
	@Resource
	private CreditServiceWlt creditServiceWlt;

	@Resource
	private TransactionTemplate transactionTemplate;

    @Resource
    private TransactionTemplate userTransactionTemplate;

	public static final String MSG_ACTIONID = "actionId";
	
	public static final String MSG_WANLIID = "wanliId";
	
	public static final String MSG_USERID = "userId";
	
    @Override
    public TopicEnum getTopicEnum() {
        return TopicEnum.USER_ADD_CREDIT;
    }


    public void doLogErrorConsumeMessage(final MsgObj msgObj) {
        List<Serializable> msgs=msgObj.getMsgList();
        for (final Serializable obj : msgs) {
            if (obj instanceof String) {
                userTransactionTemplate.execute(new TransactionCallbackWithoutResult() {
                    public void doInTransactionWithoutResult(TransactionStatus status) {
                        try {
                            HashMap<String, Object> msg = (HashMap<String, Object>)JSON.parseObject((String)obj,HashMap.class);

                            String actionId = (String) msg.get(MSG_ACTIONID);
                            RedoMonitor creditError = new RedoMonitor();
                            creditError.setId(UUID.randomUUID().toString());
                            creditError.setBusinessId(actionId);
                            creditError.setDescription("CreditAdd Error");
                            creditError.setErrorMsg(msgObj.getErrorMsg());
                            creditError.setMsgBody(obj.toString());
                            creditError.setStatus(Byte.valueOf("1"));
                            creditError.setTopic(getTopicEnum().getTopic());
                            creditErrorMapper.insertRedoMonitor(creditError);
                        } catch (Exception e) {
                            log.error("doLogErrorConsumeMessage failed", e);
                            status.setRollbackOnly();
                        } catch (Throwable t){
                            log.error("throwable catched", t);
                            status.setRollbackOnly();
                        }
                    }
                });
            }
        }
    }

    /**
     * 子类实现的，用于实际消费的代码
     *
     * @param msgObj 消息列表
     * @return 消费状态
     */
    @Override
    public ConsumeConcurrentlyStatus doConsumeMessage(final MsgObj msgObj) {
        final int maxRetryCount=super.maxRetryCount;
        List<Serializable> msgs=msgObj.getMsgList();
        for (final Serializable obj : msgs) {
            if (obj instanceof String) {
                return transactionTemplate.execute(new TransactionCallback<ConsumeConcurrentlyStatus>() {
                    public ConsumeConcurrentlyStatus doInTransaction(TransactionStatus status) {
                        try {
                            HashMap<String, Object> msg = (HashMap<String, Object>)JSON.parseObject((String)obj,HashMap.class);

                            String actionId = (String) msg.get(MSG_ACTIONID);
                            String wanliId = (String) msg.get(MSG_WANLIID);
                            long userId = (Long) msg.get(MSG_USERID);

                            if (actionId == null || wanliId == null) {
                                log.error("msg do not contains actionId or wanliId");
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            }
                            
                            Map<String, Object> selectMap = new HashMap<String, Object>();
                            selectMap.put("sync", CreditStatusEnum.hasWltNoSync.value());
                            selectMap.put("actionId", actionId);
                            selectMap.put("userId", userId);
                            CreditHisDO creditHisDO = creditHisMapper
                                .getCreditByActionIdAndSync(selectMap);
                            if (creditHisDO == null) {
                                log.error("cannot get creditHisDO by actionId and sync");
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; 
                            }

                            creditServiceWlt.syncWltCredit(wanliId, creditHisDO.getCredit(), actionId);

                            HashMap<String, Object> updateMap = new HashMap<String, Object>();
                            updateMap.put("sync", CreditStatusEnum.Synchronized.value());
                            updateMap.put("actionId", actionId);
                            updateMap.put("userId", userId);
                            updateMap.put("syncPre", CreditStatusEnum.hasWltNoSync.value());

                            creditHisMapper.updateCreditSyncByActionIdAndSync(updateMap);
                            int pointsLeft=creditServiceWlt.queryWltCredit(wanliId);
                            Map<String, Object> updatePointsMap = new HashMap<String, Object>();
                            updatePointsMap.put("id", creditHisDO.getUserId());
                            updatePointsMap.put("wanliCredits", pointsLeft);
                            userMapper.updateWanliCreditsById(updatePointsMap);
                            log.info("conSume CreitAdd Success {},wltPointsLeft {}",actionId,pointsLeft);
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        } catch (Throwable e){
                            log.error("add credit to wlt failed", e);
                            status.setRollbackOnly();
                            if(msgObj.getReconsumeTimes()>=maxRetryCount){
                                log.error("reconsumeTimes >" + maxRetryCount + "msgs:" + msgObj + "context:" + msgObj.getContext());
                                msgObj.setErrorMsg(e.getMessage());
                                doLogErrorConsumeMessage(msgObj);
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            }else{
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }



                        }
                    }
                });
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}